package xkafka

import (
	"gitee.com/go-mid/infra/xmq/mqconf"
	"gitee.com/go-mid/infra/xstat/xmetric/xprometheus"
	"gitee.com/go-mid/infra/xtrace"
	"github.com/segmentio/kafka-go"
)

func statKafkaReader(topic string, stats *kafka.ReaderStats) {
	command := "reader"
	mqconf.GetMessageNumberMetric().With("topic", topic, "command", command,
		xprometheus.LabelMessageBusType, mqconf.TraceMessageBusTypeKafka, xprometheus.LabelCallerService, xtrace.ServiceName()).Add(float64(stats.Messages))
	mqconf.GetFetchNumberMetric().With("topic", topic, "command", command,
		xprometheus.LabelMessageBusType, mqconf.TraceMessageBusTypeKafka, xprometheus.LabelCallerService, xtrace.ServiceName()).Add(float64(stats.Fetches))
	mqconf.GetTimeoutsNumberMetric().With("topic", topic, "command", command,
		xprometheus.LabelMessageBusType, mqconf.TraceMessageBusTypeKafka, xprometheus.LabelCallerService, xtrace.ServiceName()).Add(float64(stats.Timeouts))
	mqconf.GetBytesNumberMetric().With("topic", topic, "command", command,
		xprometheus.LabelMessageBusType, mqconf.TraceMessageBusTypeKafka, xprometheus.LabelCallerService, xtrace.ServiceName()).Add(float64(stats.Bytes))
	mqconf.GetErrorNumberMetric().With("topic", topic, "command", command,
		xprometheus.LabelMessageBusType, mqconf.TraceMessageBusTypeKafka, xprometheus.LabelCallerService, xtrace.ServiceName()).Add(float64(stats.Errors))
}

func statKafkaWriter(topic string, stats *kafka.WriterStats) {
	command := "writer"
	mqconf.GetMessageNumberMetric().With("topic", topic, "command", command,
		xprometheus.LabelMessageBusType, mqconf.TraceMessageBusTypeKafka, xprometheus.LabelCallerService, xtrace.ServiceName()).Add(float64(stats.Messages))
	mqconf.GetWriteNumberMetric().With("topic", topic, "command", command,
		xprometheus.LabelMessageBusType, mqconf.TraceMessageBusTypeKafka, xprometheus.LabelCallerService, xtrace.ServiceName()).Add(float64(stats.Writes))
	mqconf.GetBytesNumberMetric().With("topic", topic, "command", command,
		xprometheus.LabelMessageBusType, mqconf.TraceMessageBusTypeKafka, xprometheus.LabelCallerService, xtrace.ServiceName()).Add(float64(stats.Bytes))
	mqconf.GetErrorNumberMetric().With("topic", topic, "command", command,
		xprometheus.LabelMessageBusType, mqconf.TraceMessageBusTypeKafka, xprometheus.LabelCallerService, xtrace.ServiceName()).Add(float64(stats.Errors))
}
